package main

import (
	"context"
	"fmt"
	"github.com/segmentio/kafka-go"
	"os"
	"os/signal"
	"syscall"
	"time"
)

var r *kafka.Reader

func main() {
	r = kafka.NewReader(kafka.ReaderConfig{
		Topic:          "test-mch",
		Brokers:        []string{"localhost:9092"},
		CommitInterval: time.Second,
		StartOffset:    kafka.LastOffset,
		GroupID:        "test-group",
	})
	go si()
	for {
		m, _ := r.ReadMessage(context.Background())
		fmt.Printf("topic=%s,partition=%d,offset=%d,key=%s,value=%s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
	}

}

func si() {
	c := make(chan os.Signal, 1)
	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
	ss := <-c
	fmt.Println("get s :", ss)
	r.Close()
	os.Exit(0)
}
